/*
 * Copyright 2013-2023 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.kiki.stack.http.feign.SpringMvcContract.support;

import io.kiki.stack.http.feign.RequestTemplate;
import io.kiki.stack.http.feign.SpringMvcContract.encoding.HttpEncoding;
import io.kiki.stack.http.feign.codec.EncodeException;
import io.kiki.stack.http.feign.codec.Encoder;
import io.kiki.stack.http.feign.form.spring.SpringFormEncoder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.http.HttpMessageConverters;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpOutputMessage;
import org.springframework.http.MediaType;
import org.springframework.http.converter.ByteArrayHttpMessageConverter;
import org.springframework.http.converter.GenericHttpMessageConverter;
import org.springframework.http.converter.HttpMessageConversionException;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.http.converter.protobuf.ProtobufHttpMessageConverter;
import org.springframework.web.multipart.MultipartFile;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Stream;

import static io.kiki.stack.http.feign.SpringMvcContract.support.FeignUtils.getHttpHeaders;
import static org.springframework.http.MediaType.*;


@SuppressWarnings("rawtypes")
public class SpringEncoder implements Encoder {

    private static final Log log = LogFactory.getLog(SpringEncoder.class);

    private final SpringFormEncoder springFormEncoder;

    private final ObjectFactory<HttpMessageConverters> messageConverters;

    private final FeignEncoderProperties encoderProperties;

    private final ObjectProvider<HttpMessageConverterCustomizer> customizers;

    private List<HttpMessageConverter<?>> converters;

    public SpringEncoder(ObjectFactory<HttpMessageConverters> messageConverters) {
        this(new SpringFormEncoder(), messageConverters, new FeignEncoderProperties(), new EmptyObjectProvider<>());
    }

    public SpringEncoder(SpringFormEncoder springFormEncoder, ObjectFactory<HttpMessageConverters> messageConverters, FeignEncoderProperties encoderProperties, ObjectProvider<HttpMessageConverterCustomizer> customizers) {
        this.springFormEncoder = springFormEncoder;
        this.messageConverters = messageConverters;
        this.encoderProperties = encoderProperties;
        this.customizers = customizers;
    }

    @Override
    public void encode(Object requestBody, Type bodyType, RequestTemplate requestTemplate) throws EncodeException {
        // template.body(conversionService.convert(object, String.class));
        if (requestBody != null) {
            Collection<String> contentTypes = requestTemplate.headers().get(HttpEncoding.CONTENT_TYPE);

            MediaType requestContentType = null;
            if (contentTypes != null && !contentTypes.isEmpty()) {
                String type = contentTypes.iterator().next();
                requestContentType = MediaType.valueOf(type);
            }

            if (isFormRelatedContentType(requestContentType)) {
                springFormEncoder.encode(requestBody, bodyType, requestTemplate);
                return;
            } else {
                if (bodyType == MultipartFile.class) {
                    log.warn("For MultipartFile to be handled correctly, the 'consumes' parameter of @RequestMapping " + "should be specified as MediaType.MULTIPART_FORM_DATA_VALUE");
                }
            }
            encodeWithMessageConverter(requestBody, bodyType, requestTemplate, requestContentType);
        }
    }

    private void encodeWithMessageConverter(Object requestBody, Type bodyType, RequestTemplate request, MediaType requestContentType) {
        initConvertersIfRequired();
        for (HttpMessageConverter messageConverter : converters) {
            FeignOutputMessage outputMessage;
            try {
                if (messageConverter instanceof GenericHttpMessageConverter) {
                    outputMessage = checkAndWrite(requestBody, bodyType, requestContentType, (GenericHttpMessageConverter) messageConverter, request);
                } else {
                    outputMessage = checkAndWrite(requestBody, requestContentType, messageConverter, request);
                }
            } catch (IOException | HttpMessageConversionException ex) {
                throw new EncodeException("Error converting request body", ex);
            }
            if (outputMessage != null) {
                // clear headers
                request.headers(null);
                // converters can modify headers, so update the request
                // with the modified headers
                request.headers(new LinkedHashMap<>(outputMessage.getHeaders()));

                // do not use charset for binary data and protobuf
                Charset charset;

                MediaType contentType = outputMessage.getHeaders().getContentType();
                Charset charsetFromContentType = contentType != null ? contentType.getCharset() : null;

                if (encoderProperties != null && encoderProperties.isCharsetFromContentType() && charsetFromContentType != null) {
                    charset = charsetFromContentType;
                } else
                    if (shouldHaveNullCharset(messageConverter, outputMessage)) {
                        charset = null;
                    } else {
                        charset = StandardCharsets.UTF_8;
                    }
                request.body(outputMessage.getOutputStream().toByteArray(), charset);
                return;
            }
        }
        String message = "Could not write request: no suitable HttpMessageConverter " + "found for request type [" + requestBody.getClass().getName() + "]";
        if (requestContentType != null) {
            message += " and content type [" + requestContentType + "]";
        }
        throw new EncodeException(message);
    }

    private void initConvertersIfRequired() {
        if (converters == null) {
            converters = messageConverters.getObject().getConverters();
            customizers.forEach(customizer -> customizer.accept(converters));
        }
    }

    private boolean shouldHaveNullCharset(HttpMessageConverter messageConverter, FeignOutputMessage outputMessage) {
        return binaryContentType(outputMessage) || messageConverter instanceof ByteArrayHttpMessageConverter || messageConverter instanceof ProtobufHttpMessageConverter && ProtobufHttpMessageConverter.PROTOBUF.isCompatibleWith(outputMessage.getHeaders().getContentType());
    }

    @SuppressWarnings("unchecked")
    private FeignOutputMessage checkAndWrite(Object body, MediaType contentType, HttpMessageConverter converter, RequestTemplate request) throws IOException {
        if (converter.canWrite(body.getClass(), contentType)) {
            logBeforeWrite(body, contentType, converter);
            FeignOutputMessage outputMessage = new FeignOutputMessage(request);
            converter.write(body, contentType, outputMessage);
            return outputMessage;
        } else {
            return null;
        }
    }

    @SuppressWarnings("unchecked")
    private FeignOutputMessage checkAndWrite(Object body, Type genericType, MediaType contentType, GenericHttpMessageConverter converter, RequestTemplate request) throws IOException {
        if (converter.canWrite(genericType, body.getClass(), contentType)) {
            logBeforeWrite(body, contentType, converter);
            FeignOutputMessage outputMessage = new FeignOutputMessage(request);
            converter.write(body, genericType, contentType, outputMessage);
            return outputMessage;
        } else {
            return null;
        }
    }

    private void logBeforeWrite(Object requestBody, MediaType requestContentType, HttpMessageConverter messageConverter) {
        if (log.isDebugEnabled()) {
            if (requestContentType != null) {
                log.debug("Writing [" + requestBody + "] as \"" + requestContentType + "\" using [" + messageConverter + "]");
            } else {
                log.debug("Writing [" + requestBody + "] using [" + messageConverter + "]");
            }
        }
    }

    private boolean isFormRelatedContentType(MediaType requestContentType) {
        return isMultipartType(requestContentType) || isFormUrlEncoded(requestContentType);
    }

    private boolean isMultipartType(MediaType requestContentType) {
        return Arrays.asList(MULTIPART_FORM_DATA, MULTIPART_MIXED, MULTIPART_RELATED).contains(requestContentType);
    }

    private boolean isFormUrlEncoded(MediaType requestContentType) {
        return Objects.equals(APPLICATION_FORM_URLENCODED, requestContentType);
    }

    protected boolean binaryContentType(FeignOutputMessage outputMessage) {
        MediaType contentType = outputMessage.getHeaders().getContentType();
        return contentType == null || Stream.of(MediaType.APPLICATION_CBOR, MediaType.APPLICATION_OCTET_STREAM, MediaType.APPLICATION_PDF, MediaType.IMAGE_GIF, MediaType.IMAGE_JPEG, MediaType.IMAGE_PNG).anyMatch(mediaType -> mediaType.includes(contentType));
    }

    protected final class FeignOutputMessage implements HttpOutputMessage {

        private final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();

        private final HttpHeaders httpHeaders;

        private FeignOutputMessage(RequestTemplate request) {
            httpHeaders = getHttpHeaders(request.headers());
        }

        @Override
        public OutputStream getBody() {
            return outputStream;
        }

        @Override
        public HttpHeaders getHeaders() {
            return httpHeaders;
        }

        public ByteArrayOutputStream getOutputStream() {
            return outputStream;
        }

    }

}
